package com.mango.ch12;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;

import java.io.*;
import java.util.ArrayList;
import java.util.List;
import java.util.Vector;

/**
 * com.mango.ch12.KmeansJob
 * public：可以被所有其他类所访问
 * private：只能被本类自己访问和修改
 * protected：自身、子类及同一个包中类可以访问
 * default：同一包中的类可以访问，声明时没有加修饰符，认为是friendly。
 */
public class KmeansJob extends Configured implements Tool {
    //    private static Logger logger = Logger.getLogger(KmeansJob.class);
    private static Logger logger = Logger.getRootLogger();

    private static String intCenterFilePath = "d:/HadoopData/KMeans/Cluster.txt";
    private static String newCenterFilePath = "d:/HadoopData/KMeans/newCluster.txt";
    //    private static String intCenterFilePath = "/root/Desktop/Cluster.txt";
    private Path inputPath = new Path("d:/HadoopData/kMeans/input");
    //    private Path inputPath = new Path("/kmeans");
    private Path outputPath = new Path("d:/HadoopData/kMeans");
    //    private Path outputPath = new Path("/kmeans/output");
    private Vector<Double> lastCenters = null;
    private String groupK = "5";

    public static void main(String[] args) throws IOException {
        logger.setLevel(Level.toLevel("warn"));
        String jarPath = "C:/Users/Mango/IdeaProjects/HadoopClusterDemos/out/artifacts/HadoopDemo.jar";
        try {
            int res = ToolRunner.run(new Configuration(), new KmeansJob(), args);
            System.exit(res);
        } catch (Exception e) {
            logger.error("Job 运行出错");
            e.printStackTrace();
        }
    }


    @Override
    public int run(String[] args) throws Exception {
        Configuration conf = getConf();
        //设置新的簇点的存储的路径
        conf.set("CenterFilePath", intCenterFilePath);
        conf.set("groupK", groupK);
        //如果在Job对象获取实例之后 再修改了conf对象，则不会生效
        Job job = Job.getInstance(conf);
        job.setJarByClass(KmeansJob.class);
//        job.setCombinerClass(KmeansCombine.class);
        job.setMapperClass(KmeansMapper.class);
        job.setReducerClass(KmeansReducer.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        job.setOutputFormatClass(MyOutputFormat.class);
        FileInputFormat.addInputPath(job, inputPath);
        FileOutputFormat.setOutputPath(job, outputPath);

        job.waitForCompletion(true);
        if (job.isComplete()) logger.warn("簇点初始工作已完成");

        //循环迭代mapreduce
        long pcost = 0;
        int runTimes = 0;
        while (runTimes < 100) {
            Configuration configuration = new Configuration();
            configuration.set("CenterFilePath", newCenterFilePath);
            configuration.set("groupK", groupK);
            configuration.setFloat("lapse", (float) 1);
            Job job2 = Job.getInstance(configuration);
            job2.setMapperClass(ReKmeansMapper.class);
            job2.setReducerClass(ReKmeansReducer.class);
            job2.setMapOutputKeyClass(Text.class);
            job2.setMapOutputValueClass(Text.class);

            job2.setOutputKeyClass(Text.class);
            job2.setOutputValueClass(Text.class);

            job2.setOutputFormatClass(MyOutputFormat.class);
            FileInputFormat.addInputPath(job2, inputPath);
            FileOutputFormat.setOutputPath(job2, outputPath);
            job2.setJarByClass(this.getClass());
            job2.waitForCompletion(true);
            //获取停止迭代标志位的值  判断是否已经满足条件
            logger.warn("开始判断是否满足 停止条件 ");
//            job2.monitorAndPrintJob();
            long value = job2.getCounters().getGroup("J").findCounter("J").getValue();
            pcost = job2.getCounters().getGroup("J").findCounter("cost").getValue();

            logger.error("Counter 的 值是  " + value);
            if (value == -1) {
                break;
            }
            runTimes += 1;
        }
        logger.warn("经过" + runTimes + "的迭代计算，得到最终符合条件的簇点   \n" + " pcost：" + pcost);
        return 0;
    }

    static class KmeansMapper extends Mapper<LongWritable, Text, Text, Text> {
        private List<Vector<Double>> centers = null;
        private Path sequenceFile = null;
        private int k = 0;

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            Vector<Double> nearest = getCluster(value, centers);
            //Vector 是一个集合 double类型
            //将距离最近的簇最为键，值为本身所有字段
            context.write(new Text(vectorToString(nearest)), value);

        }

        /**
         * 读取初始化的centers
         *
         * @param context
         * @throws IOException
         * @throws InterruptedException
         */
        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            k = Integer.valueOf(context.getConfiguration().get("groupK", "5"));
            sequenceFile = new Path(context.getConfiguration().get("CenterFilePath"));
            centers = readCentersFromsequenceFile(sequenceFile, k, null);
            logger.warn(" KmeansMapper centers 中有 " + centers.size());
        }

    }

    static class ReKmeansMapper extends Mapper<LongWritable, Text, Text, Text> {
        private List<Vector<Double>> centers = null;
        private int k = 0;

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            Vector<Double> nearest = getCluster(value, centers); //Vector 是一个集合 double类型
            //将距离最近的簇最为键，值为本身所有字段
            context.write(new Text(vectorToString(nearest)), value);
        }

        /**
         * 读取初始化的centers
         *
         * @param context
         * @throws IOException
         * @throws InterruptedException
         */
        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            k = Integer.valueOf(context.getConfiguration().get("groupK", "5"));
//            logger.warn("ReKmeansMapper 从配置文件中获取到的值k是 " + k);
            String filePath = context.getConfiguration().get("CenterFilePath");
            this.centers = readCentersFromsequenceFile(new Path(filePath), k, "-->");
            //读取完所有的簇点之后将文件中内容清空 方便下次写入
            Thread.sleep(1000);
            FileOutputStream testfile = new FileOutputStream(filePath);
            testfile.write(new String("").getBytes());
            testfile.flush();
            testfile.close();
//            logger.warn(" 从newCluster文件中一共初始化了 " + centers.size() + " 个簇点");
        }
    }


    /**
     * 根据key  reduce 一下，统计一下本地的数据 做一些逻辑操作，减轻最后的Reduce的工作量
     */
    static class KmeansCombine extends Reducer<Text, Text, Text, Text> {
        private int times = 0;

        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            // key 是质心  value 是一个向量列表
            //sum 向量中的所有维都初始化为0.00
            Vector<Double> sum = new Vector<>();
            int count = 0;
            for (Text value :
                    values) {
                Vector<Double> val = strToVector(value.toString());
                //第一次的时候 sum是空的，所以直接值集里的第一个value 赋值给sum
                if (sum.isEmpty()) {
                    sum = val;
                    continue;
                }
                for (int i = 0; i < val.size(); i++) {
                    Double sums = sum.get(i);
                    sums += val.get(i);
                    sum.set(i, sums);

                }
                count += 1;
            }
            context.write(key, new Text(vectorToString(sum)));
            times++;
        }

        @Override
        protected void cleanup(Context context) throws IOException, InterruptedException {
        }
    }

    static class KmeansReducer extends Reducer<Text, Text, Text, Text> {
        int times = 0;

        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            // key 是质心  value 是一个向量列表
            //sum 向量中的所有维都初始化为0.0
            Vector<Double> newCenter = getNewCluster(values);
            String strValue = vectorToString(newCenter);
            context.write(new Text(key.toString()), new Text(strValue));
            times++;
        }

        @Override
        protected void cleanup(Context context) throws IOException, InterruptedException {
            logger.warn("KeReducer 中 centers 有 " + times);
        }
    }

    static class ReKmeansReducer extends Reducer<Text, Text, Text, Text> {
        //        private List<Vector<Double>> currCenters = null;
        //判断停止迭代的簇点误差值
        private double lapse = 0;
        private int times = 0;
        private int rd = 0;

        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            // key 是质心  value 是一个向量列表
            //sum 向量中的所有维都初始化为0.0
            Vector<Double> oldCenter = strToVector(key.toString());
            Vector<Double> newCenter = new Vector<>();
            int count = 0;
            //重新计算新的簇点
            for (Text vals :
                    values) {
                times += 1;
                count++;
                Vector<Double> val = strToVector(vals.toString());
                if (newCenter.isEmpty()) {
                    newCenter = val;
                    continue;
                }
                for (int i = 0; i < val.size(); i++) {
                    Double sum = newCenter.get(i);
                    sum += val.get(i);
                    newCenter.set(i, sum);
                }
            }
            double sum = 0;
            for (int i = 0; i < newCenter.size(); i++) {
                double newIValue = (newCenter.get(i) / count);//某维的新的均值点
                double oldIValue = oldCenter.get(i);
                newCenter.set(i, newIValue);
                sum += (newIValue - oldIValue) * (newIValue - oldIValue);
            }
            lapse += Math.sqrt(sum);
            String strValue = vectorToString(newCenter);
            context.write(new Text(key.toString()), new Text(strValue));
            rd += 1;

        }


        @Override
        protected void cleanup(Context context) throws IOException, InterruptedException {
            Thread.sleep(2000);
//            logger.warn("ReKmeansReducer reduce中一共处理了 " + currCenters.size() + " 个 簇点");
            //判断新产生的簇点是否符合 我们要求，若符合停止mapreduce的迭代
            double tmp = lapse;
            logger.warn("ReKmeansReducer 共处理了 数据 " + times + "  执行了reduce " + rd + " 次");
            logger.warn("lapse 的值是 " + lapse);
            if (lapse < context.getConfiguration().getFloat("lapse", (float) 0.5)) {
                context.getCounter("J", "J").setValue(-1);
                context.getCounter("J", "cost").setValue((long) lapse);
            } else
                context.getCounter("J", "J").setValue(1);
        }
    }

    private static List<Vector<Double>> readCentersFromsequenceFile(Path sequenceFile, int k, String regex) {
        //从SequnceFile read the centers
        //这个文件中存储的是 key  簇的中心点的值 value 是属于该簇中的所有点的集合
        List<Vector<Double>> centers = new ArrayList<>();
        BufferedReader br = null;
        try {
            br = new BufferedReader(new FileReader(sequenceFile.toString()));
            String line = null;
            while ((line = br.readLine()) != null && k != 0) {
                String[] tok = null;
                Vector<Double> tmpVec = new Vector<>();
                if (regex != null) {
                    String[] toks = line.split(regex);
                    tok = toks[1].split(",");
                } else
                    tok = line.split(",");
                for (int i = 0; i < tok.length; i++) {
                    tmpVec.add(Double.valueOf(tok[i]));
                }
                centers.add(tmpVec);
                k--;
            }
        } catch (FileNotFoundException e) {
            System.out.println("初始化的InitCenterFile 找不到，请确认路径是否正确");
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (br != null) {
                try {
                    br.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
/*        for (Vector<Double> list :
                centers) {
//            System.out.println("into readCentersFromsequenceFile  "+list.size());
            for (Double v :
                    list) {
                System.out.print(v + " ");
            }
            System.out.println();
        }*/

        return centers;
    }

    private static double[] changeTodouble(Double[] arr) {
        double[] tmp = new double[arr.length];
        for (int i = 0; i < arr.length; i++) {
            tmp[i] = arr[i];
        }
        return tmp;
    }

    private static double[] strTodouble(String values) {
        String[] tokens = values.split(",");
        double[] vals = new double[tokens.length];
        for (int i = 0; i < tokens.length; i++) {
            vals[i] = Double.valueOf(tokens[i]);
        }
        return vals;
    }

    private static double[] strTodouble(String values, String regex) {
        String[] tokens = values.split(regex);
        double[] vals = new double[tokens.length];
        for (int i = 0; i < tokens.length; i++) {
            vals[i] = Double.valueOf(tokens[i]);
        }
        return vals;
    }

    private static Vector<Double> strToVector(String values) {
        Vector<Double> vals = new Vector<>();
        String[] tokens = values.split(",");
        for (int i = 0; i < tokens.length; i++) {
            vals.add(Double.valueOf(tokens[i]));
        }
        return vals;
    }

    private static String vectorToString(Vector<Double> vector) {
        StringBuilder sb = new StringBuilder();
        for (Double vec : vector
        ) {
            sb.append(vec);
            sb.append(",");
        }
        sb.deleteCharAt(sb.lastIndexOf(","));
        return sb.toString();
    }

    /**
     * 使用欧几里德方法计算距离
     *
     * @param cValues
     * @param values
     * @return
     */
    private static double getDistance(double[] cValues, double[] values) {
        double dis = 0;
        for (int i = 0; i < values.length; i++) {
            dis += (cValues[i] - values[i]) * (cValues[i] - values[i]);
        }
        dis = Math.sqrt(dis);
        return dis;
    }

    /**
     * 获取某个point的所属簇点
     *
     * @param value
     * @param centers
     * @return
     */
    private static Vector<Double> getCluster(Text value, List<Vector<Double>> centers) {
        Vector<Double> nearest = null;
        //       传入的数据格式     1,3,2787,1698,2510,65,477,52
        //将String 数据转换为double的数据
        double[] values = strTodouble(value.toString());//获取一个d维的数据 拆分出来
        double nearestDistance = Double.MAX_VALUE;//默认值设为最大  我们要最终要取最小的distance 然后将该point加到该簇里边去
        for (Vector<Double> center : centers) {//遍历所有初始的簇，然后依次与读取进来的point 比较，将读取进来的point分组到距离最小的簇中去
            Double[] cValues = center.toArray(new Double[center.size()]);
            //使用欧几里德方法 计算距离
            double distance = getDistance(changeTodouble(cValues), values);
            if (nearest == null) {
                nearest = center;
                if (nearestDistance > distance) {
                    nearestDistance = distance;
                }
            } else {
                if (nearestDistance > distance) {
                    nearest = center;
                    nearestDistance = distance;
                }
            }
        }
        return nearest;
    }

    private static Vector<Double> getNewCluster(Iterable<Text> values) {
        Vector<Double> newCenter = new Vector<>();
        int count = 0;
        for (Text vals :
                values) {
            count++;
            Vector<Double> val = strToVector(vals.toString());
            if (newCenter.isEmpty()) {
                newCenter = val;
                continue;
            }
            for (int i = 0; i < val.size(); i++) {
                Double sum = newCenter.get(i);
                sum += val.get(i);
                newCenter.set(i, sum);
            }
        }
//            Vector<Double> vKey = strToVector(key.toString());
        for (int i = 0; i < newCenter.size(); i++) {
            newCenter.set(i, (newCenter.get(i) / count));

        }
        return newCenter;
    }
}
